/*
 * Copyright 2009 Armando Blancas
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package taskgraph.tasks.data;

import java.io.EOFException;
import java.io.IOException;
import java.io.InterruptedIOException;

import taskgraph.ports.InputPort;
import taskgraph.tasks.PrimitiveTask;


/**
 * Base implementation of a sink task.
 * It reads objects of some kind and does the work as specified in
 * subclasses. The objective is to write sink tasks with minimal effort.
 * Subclasses redefine method {@code workWith()}, adding any
 * other necessary code, in order to customize this class.
 * 
 * @author Armando Blancas
 * @see taskgraph.ports.InputPort
 * @param <E> The type of elements processed by this task.
 */
public class SinkAdapter<E> extends PrimitiveTask {

	private InputPort<E> input;
	
	/**
	 * This constructor allows creating instances as beans.
	 */
    public SinkAdapter() {
    	// do nothing
    }

	/**
     * Creates a configured instance.
     * 
     * @param input The input port.
     */
    public SinkAdapter(final InputPort<E> input) {
    	setInput(input);
    }

    /**
     * Gets the input port.
     * 
	 * @return The element input port.
	 */
	public InputPort<E> getInput() {
		return input;
	}
	
	/**
     * Performs the work of this task.
     * 
	 *<pre>
	 *while not EOF
	 *    read element E
	 *    call workWith(element)
	 *close input port
	 *</pre>
	 */
	@Override
    public void run() {
    	try {
			E element = input.read();
			while (!Thread.currentThread().isInterrupted()) {
				workWith(element);
        		element = input.read();
			}
        } catch (InterruptedIOException ie) {
        	// interruption request
        } catch (EOFException e) {
        	// we're done
        } catch (IOException e) {
        	// pipe synch error
            log.error("SinkAdapter#run()", e);
		} finally {
			input.close();
        }
    }
	
	/**
	 * Sets the element input port.
	 * 
	 * @param input The input port.
	 */
    public void setInput(final InputPort<E> input) {
    	if (input == null)
    		throw new IllegalArgumentException("input == null");
 		this.input = input;
	}

	/**
	 * To be redefined by subclasses to do the
	 * work for each element E passing through.
	 * 
	 * @param element An element read from the channel.
	 */
	protected void workWith(E element) {
		// does nothing by default
	}

}
